package com.poetic.login.fail.detect;

import akka.event.Logging;
import com.poetic.login.fail.detect.domain.LoginEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import sun.rmi.runtime.Log;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * <pre>
 *  CEP（Complex 实现):
 *      2s连续登录失败则报警输出
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class LoginFailDetectByCEP {
    public static void main(String[] args) throws Exception {
        String path = "/Users/lianghuikun/indigo/poetic/flink/user-behavior-anylysis/data/LoginLog.csv";
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        // 5402,83.149.11.115,success,1558430815
        SingleOutputStreamOperator<LoginEvent> loginEventStream = env.readTextFile(path)
                .map(new MapFunction<String, LoginEvent>() {
                    @Override
                    public LoginEvent map(String value) throws Exception {
                        String[] line = value.split("\\,");
                        LoginEvent event = LoginEvent.builder()
                                .userId(Long.valueOf(line[0]))
                                .ip(line[1])
                                .eventType(line[2])
                                .eventTime(Long.valueOf(line[3]))
                                .build();

                        return event;
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                            @Override
                            public long extractTimestamp(LoginEvent element, long recordTimestamp) {
                                return element.getEventTime();
                            }
                        }));

        // 定义匹配模式
        Pattern<LoginEvent, ?> pattern = Pattern.<LoginEvent>begin("begin")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value) throws Exception {
                        return value.getEventType().equals("fail");
                    }
                })
                .next("next")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value) throws Exception {
                        return value.getEventType().equals("fail");
                    }
                })
                .within(Time.seconds(2));

        KeyedStream<LoginEvent, Long> keyedStream = loginEventStream.keyBy(new KeySelector<LoginEvent, Long>() {
            @Override
            public Long getKey(LoginEvent value) throws Exception {
                return value.getUserId();
            }
        });

        // 在数据流中匹配出定义好的模式
        PatternStream<LoginEvent> patternStream = CEP.pattern(keyedStream, pattern);
        // .select 方法传入一个 pattern select function，当检测到定义好的模式序列时就会调用

        SingleOutputStreamOperator<Tuple3<Long, String, String>> loginFailDataStream = patternStream.select(new PatternSelectFunction<LoginEvent, Tuple3<Long, String, String>>() {
            @Override
            public Tuple3<Long, String, String> select(Map<String, List<LoginEvent>> pattern) throws Exception {

                LoginEvent first = pattern.getOrDefault("begin", null).iterator().next();
                LoginEvent second = pattern.getOrDefault("next", null).iterator().next();
                return Tuple3.of(second.getUserId(), second.getIp(), second.getEventType());
            }
        });

        loginFailDataStream.print();
        env.execute("login fail ");
    }
}
